package com.forever.disruptor;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.dsl.Disruptor;

@SuppressWarnings({ "unchecked", "rawtypes" })
public class DisruptorHelper {

	public static final int BUFFER_SIZE = 1024 * 8;
	public static DisruptorHelper instance;
	public static boolean inited = false;
	public RingBuffer<Object> ringBuffer;
	public BatchEventProcessor batchEventProcessor;
	public SequenceBarrier sequenceBarrier;
	public Disruptor disruptor;

	private DisruptorHelper(EventFactory factory, EventHandler eventHandler) {
		int count = Runtime.getRuntime().availableProcessors() + 2;
		Executor executor = Executors.newFixedThreadPool(count);
		disruptor = new Disruptor(factory, BUFFER_SIZE, executor);
		disruptor.handleEventsWith(eventHandler);
		disruptor.start();

		ringBuffer = disruptor.getRingBuffer();
		sequenceBarrier = ringBuffer.newBarrier();

		batchEventProcessor = new BatchEventProcessor(ringBuffer, sequenceBarrier, eventHandler);
		ringBuffer.addGatingSequences(batchEventProcessor.getSequence());
		System.out.println("init Executor:" + count + " threads!");
	}

	public RingBuffer getRingBuffer() {
		return ringBuffer;
	}

	public static DisruptorHelper getInstance(EventFactory factory, EventHandler eventHandler) {
		if (null == instance) {
			instance = new DisruptorHelper(factory, eventHandler);
		}
		inited = true;
		return instance;
	}
}